package day11

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
import org.apache.spark.{SparkConf, SparkContext}

import java.net.URI

/**
 * 第6章	DStream 编程进阶
 *
 *
 */
object Spark_Streaming_6 {

}

/**
 * 6.1 累加器和广播变量
 * 和 RDD 中的累加器和广播变量的用法完全一样，RDD中怎么用，这里就怎么用
 */
object Spark_Streaming_6_1 {

}

/**
 * 6.2	DataFrame and SQL Operations
 *
 * 你可以很容易地在流数据上使用DataFrames和SQL，你必须使用SparkContext来创建StreamingContext要用的SQLContext。
 * 此外，这一过程可以在驱动失效后重启。我们通过创建一个实例化的SQLContext单实例来实现这个工作。
 *
 * 如下图所示，我们对前例WordCount进行修改从而使用DataFrame和SQL来实现。每个RDD被转换为DataFrame，以临时表格配置并用SQL
 * 进行查询。
 */
object Spark_Streaming_6_2 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_6_2").setMaster("local[*]")
    //2.创建SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //3.消费kafka数据创建流
    val kafkaParams: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
    )

    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc = ssc,
      // 在大多数情况下，它将一致地在所有执行器上分布分区
      locationStrategy = LocationStrategies.PreferConsistent,
      consumerStrategy = ConsumerStrategies.Subscribe[String, String](topics = Set("my-bak"), kafkaParams)
    )

    // 结构转换
    val reduceDS: DStream[(String, Int)] = kafkaDS.map(consumerRecord => consumerRecord.value())
      .flatMap((_.split(" ")))
      .map((_, 1))
      .reduceByKey(_ + _)

    //3.创建SparkSession对象
    val spark: SparkSession = SparkSession.builder()
      .config(conf)
      .getOrCreate()

    import spark.implicits._

    reduceDS.foreachRDD(rdd => {
      val df: DataFrame = rdd.toDF(colNames = "word", "count")
      df.createOrReplaceTempView("words")
      spark.sql("select * from words").show()
    })

    // 启动
    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 6.3	Caching / Persistence
 *
 * 和RDDs类似，DStrean同样允许开发者将流数据保存在内存中。也就是说，在DStream上使用persist()方法将会自动把DStream中的每个RDD保存在内存中。
 * 当DStream中的数据要被多次计算时，这个非常有用(如在同样数据上的多次操作)。对于像reduceByWindow和reduceByKeyAndWindow以及基于状态的
 * updateStateByKey这种操作，保存是隐含默认的。因此，即使开发者没有调用persist()，由基于窗操作产生的DStream也会自动保存在内存中。
 */
object Spark_Streaming_6_3 {

}

/**
 * 6.4 优雅的关闭
 * 参考：https://blog.csdn.net/zwgdft/article/details/85849153
 */
object Spark_Streaming_6_4 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    //2.设置优雅的关闭
    sparkConf.set("spark.streaming.stopGracefullyOnShutdown", "true")
    //3.创建SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))
    // 从9999端口获取数据
    val receiverDS: ReceiverInputDStream[String] = ssc.socketTextStream(hostname = "hadoop102", port = 9999)
    // wordCount
    val reduceDS: DStream[(String, Int)] = receiverDS.flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)

    //4. 启动新的线程，希望在特殊的场合关闭SparkStreaming
    new Thread(new Runnable {
      override def run(): Unit = {
        while (true) {
          try {
            Thread.sleep(5000)
          } catch {
            case exception: Exception => println(exception)
          }

          // 监控HDFS文件的变化
          val fs: FileSystem = FileSystem.get(new URI("hdfs://hadoop102:8020"), new Configuration(), "atguigu")

          val state: StreamingContextState = ssc.getState()
          // 如果环境对象处于活动状态，可以进行关闭操作
          if (state == StreamingContextState.ACTIVE) {
            // 判断路径是否存在
            val flag: Boolean = fs.exists(new Path("hdfs://hadoop102:8020/stopSpark2"))
            if (flag) {
              println("----------------优雅地关闭---------------------")
              ssc.stop(stopSparkContext = true, stopGracefully = true)
              System.exit(0)
            }
          }

        }
      }
    }).start()

    reduceDS.print()

    // 启动
    ssc.start()
    ssc.awaitTermination()
  }
}